package com.iflytek.uoamp.agent.monitor.sender.biz.safe;

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.Logger;

import com.iflytek.uoamp.agent.core.sender.monitorsender.monitorinterface.IItem;
import com.iflytek.uoamp.agent.monitor.mcagent.SendClient;
import com.iflytek.uoamp.agent.monitor.sender.biz.AbsSender;
import com.iflytek.uoamp.agent.monitor.sender.config.Config;
import com.iflytek.uoamp.agent.monitor.sender.interfaces.ISender;
import com.iflytek.uoamp.agent.monitor.sender.interfaces.ISenderFactory;
import com.iflytek.uoamp.agent.monitor.sender.loadbalance.LoadbalanceManager;

public class SafeSender extends AbsSender {
	
	private ExecutorService callTimeoutPool;
	private AtomicLong threadCounter;
	private ISender client = SendClient.getInstance();
	private Future<Object> doAppend(final ISender client, final IItem e) throws Exception {
		return callTimeoutPool.submit(new Callable<Object>() {
			@Override
			public Object call() throws Exception {
				Object result = client.send(e);
				return result;
			}
		});
	}
	
	private Future<Object> doAppendBatch(final ISender client, final List<IItem> e) throws Exception {
		return callTimeoutPool.submit(new Callable<Object>() {
			@Override
			public Object call() throws Exception {
				Object result = client.sendBatch(e);
				return result;
			}
		});
	}
	
	
	@SuppressWarnings("rawtypes")
	public SafeSender(LoadbalanceManager lbManager,ISenderFactory factory, Config config) {
		super(lbManager,factory, config);
		this.threadCounter = new AtomicLong(0);
		this.callTimeoutPool = Executors.newFixedThreadPool(1,  new ThreadFactory() {
			@Override
			public Thread newThread(Runnable r) {
				Thread t = new Thread(r);
				t.setName("Safe Sender thread pool - " + String.valueOf(threadCounter.incrementAndGet()));
				Logger.getRootLogger().info("Make a new Safe Sender thread : " + t.getName());
				return t;
			}
		});
	}

	@Override
	public Object send(IItem item) {
//		ConnectionPoolManager manager = lbManager.getNextConnectManager();
//		if (null == manager) {
//			return null;
//		}

//		ISender client = null;
		try {
//			client = manager.checkout();
			if (null == client) {
				Logger.getRootLogger().error("SafeSender send request client failed.");
				return null;
			}
			Object object = doAppend(client, item).get(config.getRequestTimeOut(), TimeUnit.MILLISECONDS);
			return object;
		} catch (Exception e) {
			Logger.getRootLogger().error("SafeSender send item failed.", e);
		} finally {
//			if (null != client) {
//				manager.checkIn(client);
//			}
		}

		return null;
	}
	

	@Override
	public Object sendBatch(List<IItem> item) {
//		ConnectionPoolManager manager = lbManager.getNextConnectManager();
//		if (null == manager) {
//			return null;
//		}
//		ISender client = null;
		try {
//			client = manager.checkout();
			if(null == client){
				Logger.getRootLogger().error("SafeSender sendBatch request client failed.");
				return null;
			}
			
			Object obj = doAppendBatch(client, item).get(config.getRequestTimeOut(), TimeUnit.MILLISECONDS);
			return obj;
		} catch (Exception e) {
			Logger.getRootLogger().error("SafeSender sendBatch itemList failed.", e);
		} finally {
//			if (null != client) {
//				manager.checkIn(client);
//			}
		}
		return null;
	}
	
}
